/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.file;

/**
 * @Model cn.com.yang.file.ParallelLineFileReader
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月22日-下午9:21:07
 */
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

public class ParallelLineFileReader {

    private static final String FILE_PATH = "F:\\20gb_data.txt";
    private static final long CHUNK_SIZE = 1024 * 2;  // 每个线程处理的字节数，1024 * 2字节

//    private static final long CHUNK_SIZE = 1024 * 1024 * 10;  // 每个线程处理的字节数，10MB
    private static final String LINE_SEPARATOR = "\n";  // 行分隔符
    
    private static final AtomicLong processedLines = new AtomicLong(0);


    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 获取文件长度
        File file = new File(FILE_PATH);
        long fileLength = file.length();

        // 计算线程数
        int numThreads = (int) Math.ceil((double) fileLength / CHUNK_SIZE);
        
        // 使用ExecutorService来管理线程池
        ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
        List<Future<Void>> futures = new ArrayList<>();
        long start = System.currentTimeMillis();
        for (int i = 0; i < numThreads; i++) {
            final int threadIndex = i;
            futures.add(executorService.submit(() -> {
                try {
                    // 计算每个线程的起始字节和结束字节
                    long startByte = threadIndex * CHUNK_SIZE;
                    long endByte = Math.min(startByte + CHUNK_SIZE, fileLength);

                    // 使用RandomAccessFile来直接定位到文件的特定位置
                    RandomAccessFile raf = new RandomAccessFile(FILE_PATH, "r");
                    raf.seek(startByte);

                    StringBuilder chunkData = new StringBuilder();
                    byte[] buffer = new byte[2048]; // 每次读取2048字节

                    long bytesRead = startByte;
                    while (bytesRead < endByte) {
                        int len = raf.read(buffer);
                        if (len == -1) break;

                        chunkData.append(new String(buffer, 0, len));
                        bytesRead += len;

                        // 如果当前读取的数据中有行分隔符，则处理这一部分
                        while (chunkData.toString().contains(LINE_SEPARATOR)) {
                            int lineEnd = chunkData.indexOf(LINE_SEPARATOR);
                            String line = chunkData.substring(0, lineEnd);
                            chunkData.delete(0, lineEnd + LINE_SEPARATOR.length());
                            processedLines.incrementAndGet();
                            // 处理完整的一行
                            System.out.println("Thread " + threadIndex + " processed line: " + line);
                        }
                    }
                    raf.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                return null;
            }));
        }

        // 等待所有线程完成
        for (Future<Void> future : futures) {
            future.get();
        }
        // 关闭线程池
        executorService.shutdown();
        long end = System.currentTimeMillis()-start;
        System.out.println("processed total: " + processedLines.get()+",耗时：{"+ end+"}ms,{"+end/1000+"} s,"+"{"+end/1000/60+"} min");
    }
}
